package com.gukun.springboot.tdengine.service.backuprecover.recover;

import cn.hutool.core.date.BetweenFormatter;
import cn.hutool.core.date.DateUtil;
import com.gukun.springboot.tdengine.domain.backuprecover.BackupRecoverLogEntity;
import com.gukun.springboot.tdengine.domain.backuprecover.BackupRecoverTaskEntity;
import com.gukun.springboot.tdengine.domain.exception.util.ServiceExceptionUtil;
import com.gukun.springboot.tdengine.pojo.DBConfig;
import com.gukun.springboot.tdengine.service.TDEngineService;
import com.gukun.springboot.tdengine.service.backuprecover.BackupRecoverLogService;
import com.gukun.springboot.tdengine.service.backuprecover.BackupRecoverTaskService;
import com.gukun.springboot.tdengine.service.backuprecover.config.TDEngineConnPool;
import com.gukun.springboot.tdengine.service.backuprecover.config.ThreadPool;
import com.gukun.springboot.tdengine.service.backuprecover.entity.LogRecorder;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhang
 */
@Component
public class RecoverExecutor {

    @Autowired
    @Lazy
    private BackupRecoverLogService backupRecoverLogService;
    @Autowired
    @Lazy
    private BackupRecoverTaskService backupRecoverTaskService;
    @Autowired
    private TDEngineService tdEngineService;

    private final BlockingDeque<File> fileQueue = new LinkedBlockingDeque<>(3000000);
    private final ConcurrentHashMap<String,CompletableFuture<?>> FUTURE_MAP = new ConcurrentHashMap<>();
    public void execute(BackupRecoverTaskEntity task) {
        if(!FUTURE_MAP.isEmpty()){
            throw ServiceExceptionUtil.exception0(200,"当前有任务执行中，请稍后再试");
        }
        int operateTarget = task.getOperateTarget();
        Thread thread = switch (operateTarget){
            case 1 -> new Thread(()->{
                //数据库数据恢复
                try {
                    executeDbData(task);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            case 2 -> new Thread(()->{
                //数据库表恢复
                try {
                    executeDbTable(task);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            });
            default -> throw ServiceExceptionUtil.exception0(200,"恢复目标错误");
        };
        thread.start();
    }
    //表格总数量
    private AtomicLong tableCount;
    //已经读取的表格数量
    private AtomicLong tableNum;
    //文件总数量
    private AtomicLong fileCount;
    /**
     * 获取表格备份进度
     */
    public float getTableRecoverProgress(){
        float progress = 0;
        if(tableCount!=null){
            //说明日志中读取到了文件总数量
            if(tableCount.get() == 0){
                return 0;
            }
            progress = tableNum.get()*100/(float)tableCount.get();
        }else if(fileCount != null){
            if(fileCount.get() == 0){
                return 0;
            }
            progress = (fileCount.get() - fileQueue.size())*100/(float)fileCount.get();
        }
        return Float.parseFloat(String.format("%.2f",progress));
    }

    private AtomicLong dataNum;
    private AtomicLong dataCount;
    public float getTableDataRecoverProgress(){
        float progress = 0;
        if(dataCount!=null){
            //说明获取到总的数据量
            if(dataCount.get() == 0){
                return 0;
            }
            progress = dataNum.get() * 100 / (float)dataCount.get();
        }else if(fileCount != null){
            if(fileCount.get() == 0){
                return 0;
            }
            progress = (fileCount.get() - fileQueue.size()) * 100 / (float)fileCount.get();
        }
        return Float.parseFloat(String.format("%.2f",progress));
    }

    private void executeDbTable(BackupRecoverTaskEntity task) throws InterruptedException {
        Date startTime = new Date();
        String startTimeStr = DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss");
        LogRecorder logRecorder = new LogRecorder(task,backupRecoverLogService,startTimeStr);
        logRecorder.appendMessage("任务初始化");
        logRecorder.save();
        backupRecoverTaskService.updateTaskStatusStart(task);
        DBConfig database = task.getDbConfig();
        String dbName = task.getDbName();
        String jdbcUrl = database.getJdbcUrl();
        //备份表格恢复的时候，默认8个线程处理，配置多个线程处理的时候可能会有问题
        int threadNum = 8;
        logRecorder.appendMessage("执行任务线程数："+threadNum);
        logRecorder.save();
        HikariDataSource tdEngineConnPool = null;
        try{
            tdEngineConnPool = TDEngineConnPool.getHikariDataSource(jdbcUrl,threadNum);
        }catch (Exception e){
            logRecorder.appendMessage("获取数据库连接失败："+jdbcUrl);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        if(tdEngineConnPool == null){
            logRecorder.appendMessage("获取数据库连接失败："+jdbcUrl);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        //读取文件路径
        String inputDir = task.getFilePath();
        if(!inputDir.endsWith("/")){
            inputDir  = inputDir + "/";
        }
        logRecorder.updateFilePath(inputDir);
        logRecorder.appendMessage("数据文件路径："+inputDir);
        logRecorder.appendMessage("读取文件列表");
        File folder = new File(inputDir);
        File[] files = folder.listFiles();
        if(!folder.exists() || files == null){
            logRecorder.appendMessage("数据文件路径不存在："+inputDir);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        File logFile = null;
        for (File file : files) {
            if (file.isFile()) {
                String fileName = file.getName();
                if(fileName.endsWith("avro")){
                    fileQueue.add(file);
                }else if("log.txt".endsWith(fileName)){
                    logFile = file;
                }
            }
        }
        if(logFile != null){
            try(BufferedReader reader = new BufferedReader(new FileReader(logFile))){
                String line = reader.readLine();
                if (line != null){
                    tableCount = new AtomicLong(Integer.parseInt(line));
                    logRecorder.appendMessage("表格总数量："+tableCount.get());
                }
            }catch (Exception e){
                e.printStackTrace();
                logRecorder.appendMessage("读取备份日志文件失败,通过读取文件数量计算进度");
                fileCount = new AtomicLong(fileQueue.size());
            }
        }else{
            logRecorder.appendMessage("读取备份日志文件失败,通过读取文件数量计算进度");
            fileCount = new AtomicLong(fileQueue.size());
        }
        if(tableCount != null && tableCount.get() == 0){
            logRecorder.appendMessage("表格总数量为0，任务终止");
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            fileQueue.clear();
            fileCount = null;
            return;
        }
        logRecorder.appendMessage("数据文件列表："+fileQueue.size());
        logRecorder.appendMessage("开始恢复数据库表格");
        logRecorder.save();
        if(fileQueue.isEmpty()){
            logRecorder.appendMessage("数据文件提取为空，任务终止");
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            return;
        }
        //初始化超表
        logRecorder.appendMessage("初始化超表");
        boolean result = tdEngineService.addSuperTable(database.getHost(),database.getPort(),database.getUsername(),database.getPassword(),dbName);
        if(!result){
            logRecorder.appendMessage("初始化超表失败，任务终止");
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            return;
        }
        logRecorder.appendMessage("初始化超表完成");
        //多线程处理数据
        ThreadPoolExecutor threadPoolExecutor = ThreadPool.threadPoolExecutor(threadNum,threadNum);
        CompletableFuture<?>[] futures = new CompletableFuture[threadNum];
        CompletableFuture<Void> future = null;
        tableNum = new AtomicLong(0);
        for (int i=0;i<threadNum;i++) {
            future = CompletableFuture.runAsync(new RecoverDbTableTask(fileQueue,dbName,tableNum, tdEngineConnPool),threadPoolExecutor);
            futures[i] = future;
        }
        try{
            CompletableFuture<?> all = CompletableFuture.allOf(futures);
            FUTURE_MAP.put(task.getId(),all);
            all.get();
        }catch (CancellationException e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            logRecorder.appendMessage("数据库表格恢复任务主动停止");
            threadPoolExecutor.shutdownNow();
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }catch (Exception e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            logRecorder.appendMessage("数据库表格恢复任务异常终止："+e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        threadPoolExecutor.shutdown();
        if(threadPoolExecutor.awaitTermination(1000,TimeUnit.MILLISECONDS)){
            logRecorder.appendMessage("线程池关闭成功");
        }else{
            logRecorder.appendMessage("线程池关闭失败");
        }
        tdEngineConnPool.close();
        Date endTime = new Date();
        logRecorder.appendMessage("数据库表格恢复完成");
        logRecorder.appendMessage("恢复数据表格："+tableNum.get());
        logRecorder.appendMessage("恢复任务耗时："+DateUtil.formatBetween(startTime, endTime, BetweenFormatter.Level.MILLISECOND));
        fileQueue.clear();
        FUTURE_MAP.clear();
        logRecorder.updateEndTime(DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss"));
        if(logRecorder.isRunning()){
            logRecorder.saveSuccess();
        }else{
            logRecorder.saveFailed();
        }
        tableCount = null;
        tableNum = null;
        fileCount = null;
        backupRecoverTaskService.updateTaskStatusStop(task);
    }

    public void executeDbData(BackupRecoverTaskEntity task) throws InterruptedException {
        Date startTime = new Date();
        String startTimeStr = DateUtil.format(startTime,"yyyy-MM-dd HH:mm:ss");
        LogRecorder logRecorder = new LogRecorder(task,backupRecoverLogService,startTimeStr);
        logRecorder.appendMessage("任务初始化");
        logRecorder.save();
        backupRecoverTaskService.updateTaskStatusStart(task);
        DBConfig database = task.getDbConfig();
        String dbName = task.getDbName();
        String jdbcUrl = database.getJdbcUrl();
        int threadNum = task.getThreadNum();
        logRecorder.appendMessage("执行任务线程数："+threadNum);
        logRecorder.save();
        HikariDataSource tdEngineConnPool = null;
        try{
            tdEngineConnPool = TDEngineConnPool.getHikariDataSource(jdbcUrl,threadNum);
        }catch (Exception e){
            logRecorder.appendMessage("获取数据库连接失败："+jdbcUrl);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        if(tdEngineConnPool == null){
            logRecorder.appendMessage("获取数据库连接失败："+jdbcUrl);
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            return;
        }
        //读取文件路径
        String inputDir = task.getFilePath();
        if(!inputDir.endsWith("/")){
            inputDir  = inputDir + "/";
        }
        logRecorder.updateFilePath(inputDir);
        logRecorder.appendMessage("数据文件路径："+inputDir);
        logRecorder.appendMessage("读取文件列表");
        getAllFileList(inputDir);
        File logFile = new File(inputDir+"log.txt");
        if(logFile.exists() && logFile.isFile()){
            try(BufferedReader reader = new BufferedReader(new FileReader(logFile))){
                String line = reader.readLine();
                if (line != null){
                    dataCount = new AtomicLong(Long.parseLong(line));
                    logRecorder.appendMessage("数据总数量："+dataCount.get());
                }
            }catch (Exception e){
                e.printStackTrace();
                logRecorder.appendMessage("读取备份日志文件失败,通过读取文件数量计算进度");
                fileCount = new AtomicLong(fileQueue.size());
            }
        }else{
            logRecorder.appendMessage("读取备份日志文件失败,通过读取文件数量计算进度");
            fileCount = new AtomicLong(fileQueue.size());
        }
        if(dataCount != null && dataCount.get() == 0){
            logRecorder.appendMessage("从日志中读取数据总数量为0，任务终止");
            logRecorder.saveSuccess();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            fileQueue.clear();
            fileCount = null;
            return;
        }
        logRecorder.appendMessage("数据文件列表："+fileQueue.size());
        logRecorder.appendMessage("开始恢复数据库数据");
        logRecorder.save();
        if(fileQueue.isEmpty()){
            logRecorder.appendMessage("数据文件提取为空，任务终止");
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
            tdEngineConnPool.close();
            return;
        }
        //多线程处理数据
        ThreadPoolExecutor threadPoolExecutor = ThreadPool.threadPoolExecutor(threadNum,threadNum);
        CompletableFuture<?>[] futures = new CompletableFuture[threadNum];
        CompletableFuture<Void> future = null;
        dataNum = new AtomicLong(0);
        for (int i=0;i<threadNum;i++) {
            future = CompletableFuture.runAsync(new RecoverDbDataTask(fileQueue,dbName,dataNum, tdEngineConnPool),threadPoolExecutor);
            futures[i] = future;
        }
        try{
            CompletableFuture<?> all = CompletableFuture.allOf(futures);
            FUTURE_MAP.put(task.getId(),all);
            all.get();
        }catch (CancellationException e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            logRecorder.appendMessage("数据库数据恢复任务主动停止");
            threadPoolExecutor.shutdownNow();
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }catch (Exception e){
            e.printStackTrace();
            for(CompletableFuture<?> item : futures){
                if(item != null && !item.isCancelled() && !item.isDone()){
                    item.cancel(true);
                }
            }
            logRecorder.appendMessage("数据库数据恢复任务异常终止："+e.getMessage());
            logRecorder.saveFailed();
            backupRecoverTaskService.updateTaskStatusStop(task);
        }
        threadPoolExecutor.shutdown();
        if(threadPoolExecutor.awaitTermination(1000,TimeUnit.MILLISECONDS)){
            logRecorder.appendMessage("线程池关闭成功");
        }else{
            logRecorder.appendMessage("线程池关闭失败");
        }
        tdEngineConnPool.close();
        Date endTime = new Date();
        logRecorder.appendMessage("数据库数据恢复完成");
        logRecorder.appendMessage("恢复数据条数："+dataNum.get());
        logRecorder.appendMessage("恢复任务耗时："+DateUtil.formatBetween(startTime, endTime, BetweenFormatter.Level.MILLISECOND));
        fileQueue.clear();
        FUTURE_MAP.clear();
        fileCount = null;
        dataCount = null;
        dataNum = null;
        logRecorder.updateEndTime(DateUtil.format(endTime,"yyyy-MM-dd HH:mm:ss"));
        if(logRecorder.isRunning()){
            logRecorder.saveSuccess();
        }else{
            logRecorder.saveFailed();
        }
        backupRecoverTaskService.updateTaskStatusStop(task);
    }

    private void getAllFileList(String inputDir) {
        File file = new File(inputDir);
        if (!file.exists()) {
            throw new RuntimeException("文件不存在");
        }
        File[] files = file.listFiles();
        if(files==null){
            return;
        }
        File[] children = null;
        for(File f:files){
            if(f.isDirectory()){
                children = f.listFiles();
                if(children == null){
                    continue;
                }
                fileQueue.addAll(Arrays.asList(children));
            }
        }
    }

    public void stopTask(String id) {
        CompletableFuture<?> future = FUTURE_MAP.get(id);
        if(future != null){
            future.cancel(true);
        }else{
            throw ServiceExceptionUtil.exception0(500, "任务初始化未完成");
        }
    }
}
